package cn.com.kh;

import cn.com.elite.Util;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Arrays;
import java.util.Properties;

/**
 * @Title: cn.com.kh-kafkaUtil
 * @Description:ngsapitask-pa 系统API接口开发Demo，重点关注业务逻辑部分
 * @Author: lijun
 * @Date 2021-12-18 20:31
 */
public class kafkaUtil {
    private static final Log log = LogFactory.getLog(kafkaUtil.class);

    public static String getMsg(String servers, String groupID, String[] topic, int getLimit, String callback) {

        String resultInfo = null;
        int callTryNum = 0;
        String checkNullInfo = KHUtil.isNullMSG("servers", servers) + KHUtil.isNullMSG("groupID", groupID) + KHUtil.isNullMSG("topic", Arrays.toString(topic)) + KHUtil.isNullMSG("getLimit", String.valueOf(getLimit) + KHUtil.isNullMSG("callback", callback));
        if (!KHUtil.isNull(checkNullInfo)) {
            log.warn("非空校验检查结果:" + checkNullInfo);
            return checkNullInfo;
        }

        Properties properties = new Properties();
        properties.put("bootstrap.servers", servers);
        properties.put("group.id", groupID);
        properties.put("enable.auto.commit", "false");
        properties.put("auto.commit.interval.ms", "1000");
        properties.put("auto.offset.reset", "latest");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        consumer.subscribe(Arrays.asList(topic));

        JSONObject resultObj = new JSONObject();
        JSONArray recordArr = new JSONArray();
        try {
            while (true) {
                System.out.println("asfd");
                ConsumerRecords<String, String> records = consumer.poll(getLimit);

                for (ConsumerRecord<String, String> record : records) {

                    JSONObject recordObj = new JSONObject();
                    recordObj.put("topic", record.topic());
                    recordObj.put("offset", record.offset());
                    recordObj.put("key", record.key());
                    recordObj.put("value", record.value());
                    recordArr.add(recordObj);
                }

                if (recordArr.size() > 0) {
                    resultObj.put("code", 1);
                    resultObj.put("msg", "获取成功!");
                    resultObj.put("data", recordArr);

                    log.debug("回调地址：【" + callback + "】 回调信息：【" + resultObj.toString() + "】");

                    // 将结果推送回调地址
                    callTryNum = 1;
                    do {
                        String callResult = Util.doHttpPost(callback, resultObj.toString(), "UTF-8", 120000, 0);
                        log.debug("结果推送回调地址【" + callTryNum + "】:" + callResult);

                        JSONObject callResultJson = JSONObject.fromObject(callResult);
                        if (!KHUtil.isNull(callResult) && callResultJson.getInt("code") == 1) {
                            callTryNum = 99;
                        } else {
                            callTryNum++;
                            if (callTryNum == 1) {
                                log.warn("结果推送回调地址错误【" + callTryNum + "】:" + callResult);
                            }
                        }
                    } while (callTryNum <= 3);

                    resultObj.clear();
                    recordArr.clear();

                }
                // 异步提交
                consumer.commitAsync();
            }
        } catch (Exception e) {
            log.error("获取kafka消息异常Topic【" + Arrays.toString(topic) + "】"+e.getMessage());
            resultInfo = e.getMessage();
        } finally {
            try {
                // 因为即将要关闭消费者，所以要用同步提交保证提交成功
                consumer.commitSync();
            } finally {
                consumer.close();
            }
        }

        return resultInfo;

    }

    public static JSONObject sendMsg(String servers, String topic, String data) {
        JSONObject resultObj = new JSONObject();

        String checkNullInfo = KHUtil.isNullMSG("servers", servers) + KHUtil.isNullMSG("topic", topic) + KHUtil.isNullMSG("data", data);
        log.warn("非空校验检查结果:" + checkNullInfo);
        if (!KHUtil.isNull(checkNullInfo)) {
            resultObj.put("code", 0);
            resultObj.put("msg", checkNullInfo);

        } else {
            Properties props = new Properties();
            props.put("bootstrap.servers", servers);//xxx服务器ip
            props.put("acks", "all");//所有follower都响应了才认为消息提交成功，即"committed"
            props.put("retries", 0);//retries = MAX 无限重试，直到你意识到出现了问题:)
            props.put("batch.size", 16384);//producer将试图批处理消息记录，以减少请求次数.默认的批量处理消息字节数
            //batch.size当批量的数据大小达到设定值后，就会立即发送，不顾下面的linger.ms
            props.put("linger.ms", 1);//延迟1ms发送，这项设置将通过增加小的延迟来完成--即，不是立即发送一条记录，producer将会等待给定的延迟时间以允许其他消息记录发送，这些消息记录可以批量处理
            props.put("buffer.memory", 33554432);//producer可以用来缓存数据的内存大小。
            props.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

            KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

            try {
                producer.send(new ProducerRecord<String, String>(topic, data));
                resultObj.put("code", 1);
                resultObj.put("msg", "调用成功!");
            } catch (Exception e) {
                log.error("发送kafka消息异常Topic【" + topic + "】", e);
                resultObj.put("code", -1);
                resultObj.put("msg", e.getMessage());
            } finally {

                producer.close();
            }
        }

        return resultObj;

    }
}
